package com.networkflow_analysis

import com.networkflow_analysis.bean.{PvCount, UserBehavior, UvCount}
import org.apache.flink.api.common.functions.{AggregateFunction, MapFunction}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.function.{AllWindowFunction, WindowFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.util.Random

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/29 23:26
  *                  网站独立访客数
  *                  Uv统计，去重同用户,访问网站的总人数
  */
object UniqueVisitor {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    //读取数据，转换成样例类提取时间戳和watermark
    val resource = getClass.getResource("/").toURI
    val filePath: String = "NetworkFlowAnalysis/src/main/resources/UserBehavior.csv"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    val dataStream: DataStream[UserBehavior] = fileStream.map(data => {
      val arr = data.split(",");
      UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toInt, arr(3), arr(4).toLong)
    })
      .assignAscendingTimestamps(_.timestamp * 1000L)

    val UvStream: DataStream[UvCount] = dataStream.filter(_.behavior == "pv")
      .timeWindowAll(Time.hours(1)) //直接不分组，基于DataStream开1小时滚动窗口
      .apply(new UvCountResult)

    UvStream.print("uv job")


    env.execute("pv job")
  }

  //自定义全窗口函数，用一个Set结构来保存所有的数据，进行自动去重
  class UvCountResult extends AllWindowFunction[UserBehavior, UvCount, TimeWindow] {
    override def apply(window: TimeWindow, input: Iterable[UserBehavior], out: Collector[UvCount]): Unit = {
      var userIdSet =Set[Long]()

      //遍历窗口中所有数据，把userId添加到set中，自动去重
      for (userBehavior<-input){
        userIdSet+=userBehavior.userId
      }
      out.collect(UvCount(window.getEnd,userIdSet.size))
    }
  }


}
